#include "redis/store/redis_zsets.h"

#include "redis/store/base_data_key_format.h"
#include "redis/store/base_meta_value_format.h"
#include "redis/store/zsets_comparator.h"
#include "redis/store/zsets_data_key_format.h"

rocksdb::Comparator *ZSetsScoreKeyComparator() {
  static ZSetsComparator zsets_comaprator;
  return &zsets_comaprator;
}

rocksdb::Status RedisZSets::Open(
    rocksdb::TransactionDB *transcation_db_, rocksdb::Status s,
    std::vector<rocksdb::ColumnFamilyDescriptor> &column_families) {
  if (s.ok()) {
    rocksdb::ColumnFamilyHandle *mcf = nullptr, *dcf = nullptr, *scf = nullptr;
    s = transcation_db_->CreateColumnFamily(rocksdb::ColumnFamilyOptions(),
                                            "zsets_meta_cf", &mcf);
    if (!s.ok()) {
      return s;
    }

    s = transcation_db_->CreateColumnFamily(rocksdb::ColumnFamilyOptions(),
                                            "zsets_data_cf", &dcf);
    if (!s.ok()) {
      return s;
    }

    rocksdb::ColumnFamilyOptions score_cf_ops;
    score_cf_ops.comparator = ZSetsScoreKeyComparator();
    s = transcation_db_->CreateColumnFamily(score_cf_ops, "zsets_score_cf",
                                            &scf);
    if (!s.ok()) {
      return s;
    }

    s = transcation_db_->DestroyColumnFamilyHandle(dcf);
    if (!s.ok()) {
      return s;
    }

    s = transcation_db_->DestroyColumnFamilyHandle(scf);
    if (!s.ok()) {
      return s;
    }
    delete transcation_db_;
  }

  rocksdb::DBOptions db_ops(options_);
  rocksdb::ColumnFamilyOptions default_cf_ops(options_);
  rocksdb::ColumnFamilyOptions meta_cf_ops(options_);
  rocksdb::ColumnFamilyOptions data_cf_ops(options_);
  rocksdb::ColumnFamilyOptions score_cf_ops(options_);

  score_cf_ops.comparator = ZSetsScoreKeyComparator();
  column_families.push_back(
      rocksdb::ColumnFamilyDescriptor("zsets_meta_cf", meta_cf_ops));
  column_families.push_back(
      rocksdb::ColumnFamilyDescriptor("zsets_data_cf", data_cf_ops));
  column_families.push_back(
      rocksdb::ColumnFamilyDescriptor("zsets_score_cf", score_cf_ops));
  column_families.push_back(rocksdb::ColumnFamilyDescriptor(
      rocksdb::kDefaultColumnFamilyName, default_cf_ops));
  return rocksdb::Status::OK();
}

RedisZSets::RedisZSets(const rocksdb::Options &options,
                       std::vector<rocksdb::ColumnFamilyHandle *> &cf_handles)
    : options_(options),
      cf_handles_(cf_handles),
      scope_lock_mgr_(new ScopeLockMgr(4096)) {}

RedisZSets::~RedisZSets() {}

rocksdb::Status RedisZSets::ZAdd(
    const rocksdb::Slice &key, const std::vector<ScoreMember> &score_members,
    int32_t *ret, const std::shared_ptr<rocksdb::Transaction> &transcation) {
  *ret = 0;
  std::unordered_set<std::string> unique;
  std::vector<ScoreMember> filtered_score_members;
  for (const auto &sm : score_members) {
    if (unique.find(sm.member) == unique.end()) {
      unique.insert(sm.member);
      filtered_score_members.push_back(sm);
    }
  }

  char score_buf[8];
  int32_t version = 0;
  std::string meta_value;
  rocksdb::WriteBatch batch;
  rocksdb::Status s;
  ScopeLock l(scope_lock_mgr_.get(), key);
  if (transcation) {
    s = transcation->Get(default_read_options_, cf_handles_[0], key,
                         &meta_value);
  } else {
    s = transcation_db_->Get(default_read_options_, cf_handles_[0], key,
                             &meta_value);
  }

  if (s.ok()) {
    bool vaild = true;
    ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
    if (parsed_zsets_meta_value.IsStale() ||
        parsed_zsets_meta_value.count() == 0) {
      vaild = false;
      version = parsed_zsets_meta_value.InitialMetaValue();
    } else {
      vaild = true;
      version = parsed_zsets_meta_value.verison();
    }

    int32_t count = 0;
    std::string data_value;
    for (const auto &score_member : filtered_score_members) {
      bool not_found = true;
      ZSetsMemberKey zsets_member_key(key, version, score_member.member);
      if (vaild) {
        if (transcation) {
          s = transcation->Get(default_read_options_, cf_handles_[1],
                               zsets_member_key.Encode(), &data_value);
        } else {
          s = transcation_db_->Get(default_read_options_, cf_handles_[1],
                                   zsets_member_key.Encode(), &data_value);
        }

        if (s.ok()) {
          not_found = false;
          uint64_t tmp = slash::DecodeFixed64(data_value.data());
          const void *ptr_tmp = reinterpret_cast<const void *>(&tmp);
          double old_score = *reinterpret_cast<const double *>(ptr_tmp);
          if (old_score == score_member.score) {
            continue;
          } else {
            ZSetsScoreKey zsets_score_key(key, version, old_score,
                                          score_member.member);
            if (transcation) {
              s = transcation->Delete(cf_handles_[2], zsets_score_key.Encode());
            } else {
              s = batch.Delete(cf_handles_[2], zsets_score_key.Encode());
            }

            if (!s.ok()) {
              return s;
            }
          }
        } else if (!s.IsNotFound()) {
          return s;
        }
      }

      const void *ptr_score =
          reinterpret_cast<const void *>(&score_member.score);
      slash::EncodeFixed64(score_buf,
                           *reinterpret_cast<const uint64_t *>(ptr_score));
      if (transcation) {
        s = transcation->Put(cf_handles_[1], zsets_member_key.Encode(),
                             rocksdb::Slice(score_buf, sizeof(uint64_t)));
      } else {
        s = batch.Put(cf_handles_[1], zsets_member_key.Encode(),
                      rocksdb::Slice(score_buf, sizeof(uint64_t)));
      }

      if (!s.ok()) {
        return s;
      }

      ZSetsScoreKey zsets_score_key(key, version, score_member.score,
                                    score_member.member);
      if (transcation) {
        s = transcation->Put(cf_handles_[2], zsets_score_key.Encode(),
                             rocksdb::Slice());
      } else {
        s = batch.Put(cf_handles_[2], zsets_score_key.Encode(),
                      rocksdb::Slice());
      }
      if (not_found) {
        count++;
      }
    }

    parsed_zsets_meta_value.ModifyCount(count);
    if (transcation) {
      s = transcation->Put(cf_handles_[0], key, meta_value);
    } else {
      s = batch.Put(cf_handles_[0], key, meta_value);
    }

    if (!s.ok()) {
      return s;
    }
    *ret = count;
  } else if (s.IsNotFound()) {
    char buf[4];
    slash::EncodeFixed32(buf, filtered_score_members.size());
    ZSetsMetaValue zsets_meta_value(rocksdb::Slice(buf, sizeof(int32_t)));
    version = zsets_meta_value.UpdateVersion();
    if (transcation) {
      s = transcation->Put(cf_handles_[0], key, zsets_meta_value.Encode());
    } else {
      s = batch.Put(cf_handles_[0], key, zsets_meta_value.Encode());
    }

    if (!s.ok()) {
      return s;
    }

    for (const auto &score_member : filtered_score_members) {
      ZSetsMemberKey zsets_member_key(key, version, score_member.member);
      const void *ptr_score =
          reinterpret_cast<const void *>(&score_member.score);
      slash::EncodeFixed64(score_buf, *reinterpret_cast<const uint64_t *>(ptr_score));
      if (transcation) {
        s = transcation->Put(cf_handles_[1], zsets_member_key.Encode(),
                             rocksdb::Slice(score_buf, sizeof(uint64_t)));
      } else {
        s = batch.Put(cf_handles_[1], zsets_member_key.Encode(),
                      rocksdb::Slice(score_buf, sizeof(uint64_t)));
      }

      if (!s.ok()) {
        return s;
      }

      ZSetsScoreKey zsets_score_key(key, version, score_member.score,
                                    score_member.member);
      if (transcation) {
        s = transcation->Put(cf_handles_[2], zsets_score_key.Encode(),
                             rocksdb::Slice());
      } else {
        s = batch.Put(cf_handles_[2], zsets_score_key.Encode(),
                      rocksdb::Slice());
      }

      if (!s.ok()) {
        return s;
      }
    }
    *ret = filtered_score_members.size();
  } else {
    return s;
  }

  if (!transcation) {
    s = transcation_db_->Write(default_write_options_, &batch);
  }
  return s;
}

rocksdb::Status RedisZSets::ZCard(
    const rocksdb::Slice &key, int32_t *card,
    const std::shared_ptr<rocksdb::Transaction> &transcation) {
  *card = 0;
  std::string meta_value;
  rocksdb::Status s;
  if (transcation) {
    s = transcation->Get(default_read_options_, cf_handles_[0], key,
                         &meta_value);
  } else {
    s = transcation_db_->Get(default_read_options_, cf_handles_[0], key,
                             &meta_value);
  }

  if (s.ok()) {
    ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
    if (parsed_zsets_meta_value.IsStale()) {
      *card = 0;
      return rocksdb::Status::NotFound("Stale");
    } else if (parsed_zsets_meta_value.count() == 0) {
      *card = 0;
      return rocksdb::Status::NotFound();
    } else {
      *card = parsed_zsets_meta_value.count();
    }
  }
  return s;
}

rocksdb::Status RedisZSets::ZCount(
    const rocksdb::Slice &key, double min, double max, bool left_close,
    bool right_close, int32_t *ret,
    const std::shared_ptr<rocksdb::Transaction> &transcation) {
  *ret = 0;
  rocksdb::ReadOptions read_options;
  const rocksdb::Snapshot *snapshot = nullptr;

  std::string meta_value;
  ScopeSnapshot ss(transcation_db_.get(), &snapshot);
  read_options.snapshot = snapshot;
  rocksdb::Status s;
  if (transcation) {
    s = transcation->Get(read_options, cf_handles_[0], key, &meta_value);
  } else {
    s = transcation_db_->Get(read_options, cf_handles_[0], key, &meta_value);
  }

  if (s.ok()) {
    ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
    if (parsed_zsets_meta_value.IsStale()) {
      return rocksdb::Status::NotFound("Stale");
    } else if (parsed_zsets_meta_value.count() == 0) {
      return rocksdb::Status::NotFound();
    } else {
      int32_t version = parsed_zsets_meta_value.verison();
      int32_t cnt = 0;
      int32_t cur_index = 0;
      int32_t stop_index = parsed_zsets_meta_value.count() - 1;
      ScoreMember score_member;
      ZSetsScoreKey zsets_score_key(key, version,
                                    std::numeric_limits<double>::lowest(),
                                    rocksdb::Slice());
      std::unique_ptr<rocksdb::Iterator> iter = nullptr;
      if (transcation) {
        iter = std::unique_ptr<rocksdb::Iterator>(
            transcation->GetIterator(read_options, cf_handles_[2]));
      } else {
        iter = std::unique_ptr<rocksdb::Iterator>(
            transcation_db_->NewIterator(read_options, cf_handles_[2]));
      }

      for (iter->Seek(zsets_score_key.Encode());
           iter->Valid() && cur_index <= stop_index;
           iter->Next(), ++cur_index) {
        bool left_pass = false;
        bool right_pass = false;
        ParsedZSetsScoreKey parsed_zsets_score_key(iter->key());
        if ((left_close && min >= parsed_zsets_score_key.score()) ||
            (!left_close && min > parsed_zsets_score_key.score())) {
          left_pass = true;
        }

        if ((right_close && parsed_zsets_score_key.score() <= max) ||
            (!right_close && parsed_zsets_score_key.score() < max)) {
          right_pass = true;
        }

        if (left_pass && right_pass) {
          cnt++;
        } else if (!right_pass) {
          break;
        }
      }
      *ret = cnt;
    }
  }
  return s;
}

rocksdb::Status RedisZSets::ZIncrby(
    const rocksdb::Slice &key, const rocksdb::Slice &member, double increment,
    double *ret, const std::shared_ptr<rocksdb::Transaction> &transcation) {
  *ret = 0;
  uint32_t statistic = 0;
  double score = 0;
  char score_buf[8];
  int32_t version = 0;
  std::string meta_value;
  rocksdb::WriteBatch batch;
  ScopeLock l(scope_lock_mgr_.get(), key);
  rocksdb::Status s;
  if (transcation) {
    s = transcation->Get(default_read_options_, cf_handles_[0], key,
                         &meta_value);
  } else {
    s = transcation_db_->Get(default_read_options_, cf_handles_[0], key,
                             &meta_value);
  }

  if (s.ok()) {
    ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
    if (parsed_zsets_meta_value.IsStale() ||
        parsed_zsets_meta_value.count() == 0) {
      version = parsed_zsets_meta_value.InitialMetaValue();
    } else {
      version = parsed_zsets_meta_value.verison();
    }

    std::string data_value;
    ZSetsMemberKey zsets_member_key(key, version, member);
    if (transcation) {
      s = transcation->Get(default_read_options_, cf_handles_[1],
                           zsets_member_key.Encode(), &data_value);
    } else {
      s = transcation_db_->Get(default_read_options_, cf_handles_[1],
                               zsets_member_key.Encode(), &data_value);
    }

    if (s.ok()) {
      uint64_t tmp = slash::DecodeFixed64(data_value.data());
      const void *ptr_tmp = reinterpret_cast<const void *>(&tmp);
      double old_score = *reinterpret_cast<const double *>(ptr_tmp);
      score = old_score + increment;
      ZSetsScoreKey zsets_score_key(key, version, old_score, member);
      if (transcation) {
        transcation->Delete(cf_handles_[2], zsets_score_key.Encode());
      } else {
        batch.Delete(cf_handles_[2], zsets_score_key.Encode());
      }
    } else if (s.IsNotFound()) {
      score = increment;
      parsed_zsets_meta_value.ModifyCount(1);
      if (transcation) {
        transcation->Put(cf_handles_[0], key, meta_value);
      } else {
        batch.Put(cf_handles_[0], key, meta_value);
      }
    } else {
      return s;
    }
  } else if (s.IsNotFound()) {
    char buf[4];
    slash::EncodeFixed32(buf, 1);
    ZSetsMetaValue zsets_meta_value(rocksdb::Slice(buf, sizeof(int32_t)));
    version = zsets_meta_value.UpdateVersion();
    batch.Put(cf_handles_[0], key, zsets_meta_value.Encode());
    score = increment;
  } else {
    return s;
  }

  ZSetsMemberKey zsets_member_key(key, version, member);
  const void *ptr_score = reinterpret_cast<const void *>(&score);
  slash::EncodeFixed64(score_buf,
                       *reinterpret_cast<const uint64_t *>(ptr_score));
  if (transcation) {
    transcation->Put(cf_handles_[1], zsets_member_key.Encode(),
                     rocksdb::Slice(score_buf, sizeof(uint64_t)));
  } else {
    batch.Put(cf_handles_[1], zsets_member_key.Encode(),
              rocksdb::Slice(score_buf, sizeof(uint64_t)));
  }

  ZSetsScoreKey zsets_score_key(key, version, score, member);
  if (transcation) {
    transcation->Put(cf_handles_[2], zsets_score_key.Encode(),
                     rocksdb::Slice());
  } else {
    batch.Put(cf_handles_[2], zsets_score_key.Encode(), rocksdb::Slice());
  }

  *ret = score;
  if (!transcation) {
    s = transcation_db_->Write(default_write_options_, &batch);
  }
  return s;
}

rocksdb::Status RedisZSets::ZRange(
    const rocksdb::Slice &key, int32_t start, int32_t stop,
    std::vector<ScoreMember> *score_members,
    const std::shared_ptr<rocksdb::Transaction> &transcation) {
  score_members->clear();
  rocksdb::ReadOptions read_options;
  const rocksdb::Snapshot *snapshot = nullptr;

  std::string meta_value;
  ScopeSnapshot ss(transcation_db_.get(), &snapshot);
  read_options.snapshot = snapshot;
  rocksdb::Status s;
  if (transcation) {
    s = transcation->Get(default_read_options_, cf_handles_[0], key,
                         &meta_value);
  } else {
    s = transcation_db_->Get(default_read_options_, cf_handles_[0], key,
                             &meta_value);
  }

  if (s.ok()) {
    ParsedZSetsMetaValue parsed_zsets_meta_value(&meta_value);
    if (parsed_zsets_meta_value.IsStale()) {
      return rocksdb::Status::NotFound("Stale");
    } else if (parsed_zsets_meta_value.count() == 0) {
      return rocksdb::Status::NotFound();
    } else {
      int32_t count = parsed_zsets_meta_value.count();
      int32_t version = parsed_zsets_meta_value.verison();
      int32_t start_index = start >= 0 ? start : count + start;
      int32_t stop_index = stop >= 0 ? stop : count + stop;
      start_index = start_index <= 0 ? 0 : start_index;
      stop_index = stop_index >= count ? count - 1 : stop_index;
      if (start_index > stop_index || start_index >= count || stop_index < 0) {
        return s;
      }

      int32_t cur_index = 0;
      ZSetsScoreKey zsets_score_key(key, version,
                                    std::numeric_limits<double>::lowest(),
                                    rocksdb::Slice());
      std::unique_ptr<rocksdb::Iterator> iter = nullptr;
      if (transcation) {
        iter = std::unique_ptr<rocksdb::Iterator>(
            transcation->GetIterator(read_options, cf_handles_[2]));
      } else {
        iter = std::unique_ptr<rocksdb::Iterator>(
            transcation_db_->NewIterator(read_options, cf_handles_[2]));
      }

      for (iter->Seek(zsets_score_key.Encode());
           iter->Valid() && cur_index <= stop_index;
           iter->Next(), ++cur_index) {
        if (cur_index >= start_index) {
          ParsedZSetsScoreKey parsed_zsets_score_key(iter->key());
          ScoreMember score_member;
          score_member.score = parsed_zsets_score_key.score();
          score_member.member = parsed_zsets_score_key.member().ToString();
          score_members->emplace_back(score_member);
        }
      }
    }
  }
  return s;
}

rocksdb::Status RedisZSets::ZRank(
    const rocksdb::Slice &key, const rocksdb::Slice &member, int32_t *rank,
    const std::shared_ptr<rocksdb::Transaction> &transcation) {
  *rank = -1;
  rocksdb::ReadOptions read_options;
  const rocksdb::Snapshot *snapshot = nullptr;

  std::string meta_value;
  ScopeSnapshot ss(transcation_db_.get(), &snapshot);
  read_options.snapshot = snapshot;
  rocksdb::Status s;
  if (transcation) {
    s = transcation->Get(default_read_options_, cf_handles_[0], key,
                         &meta_value);
  } else {
    s = transcation_db_->Get(default_read_options_, cf_handles_[0], key,
                             &meta_value);
  }

  if (s.ok()) {
    ParsedZSetsMetaValue parses_sets_meta_value(&meta_value);
    if (parses_sets_meta_value.IsStale()) {
      return rocksdb::Status::NotFound("Stale");
    } else if (parses_sets_meta_value.count() == 0) {
      return rocksdb::Status::NotFound("");
    } else {
      bool found = false;
      int32_t version = parses_sets_meta_value.verison();
      int32_t index = 0;
      int32_t stop_index = parses_sets_meta_value.count() - 1;
      ScoreMember score_member;
      ZSetsScoreKey zsets_score_key(key, version,
                                    std::numeric_limits<double>::lowest(),
                                    rocksdb::Slice());
      std::unique_ptr<rocksdb::Iterator> iter = nullptr;
      if (transcation) {
        iter = std::unique_ptr<rocksdb::Iterator>(
            transcation->GetIterator(read_options, cf_handles_[2]));
      } else {
        iter = std::unique_ptr<rocksdb::Iterator>(
            transcation_db_->NewIterator(read_options, cf_handles_[2]));
      }

      for (iter->Seek(zsets_score_key.Encode());
           iter->Valid() && index <= stop_index; iter->Next(), ++index) {
        ParsedZSetsScoreKey parsed_zsets_score_key(iter->key());
        if (!parsed_zsets_score_key.member().compare(member)) {
          found = true;
          break;
        }
      }

      if (found) {
        *rank = index;
        return rocksdb::Status::OK();
      } else {
        return rocksdb::Status::NotFound();
      }
    }
  }
  return s;
}

rocksdb::Status RedisZSets::ZScan(const Slice &key, int64_t cursor,
  const std::string& pattern, int64_t count,
  std::vector<ScoreMember>* score_members, int64_t *next_cursor,
    const std::shared_ptr<rocksdb::Transaction> &transcation) {

}